package org.niit.service

import java.text.SimpleDateFormat

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.niit.common.TService
import org.niit.dao.BlackListDao
import org.niit.handler.DataHandler
import org.niit.bean.AdClickData
import org.niit.util.MyKafkaUtil

class BlackListService extends TService{



  override def dataAnalysis(data:DStream[AdClickData]): Any = {

    //在Kafka中获取广告点击数据
    val ds: DStream[((String, String, String), Int)] = filterKafkaData(data)
	  //检查用户点击次数
    checkUserCount(ds)


//    ssc.start()
//    ssc.awaitTermination()
    }

  //利用在Kafka中采集出来的流式数据对数据库进行实时检索
  //将已经在黑名单的数据过滤掉，过滤掉后存进每日用户点击广告次数表中
  private def filterKafkaData(adClickData: DStream[AdClickData]): DStream[((String, String, String), Int)] ={
      //1.将DStream转换成rdd，再执行过滤等操作
      adClickData.transform(rdd=>{

        //判断点击用户是否在黑名单当中，将在黑名单中的数据剔除（过滤掉）
        val filterRdd = rdd.filter( data =>{
          val blackListDao = new BlackListDao
          !blackListDao.selectBlackUserById(data.user)  //将不在黑名单中的数据保存出来
        })
      //filterRdd 是保存的不在黑名单中的数据
        //将 filterRdd 转换成 元组 (hello,1) (hello,1) (hello,1) ==> (hello,3)
        //  （ (日期,用户，广告) ，1）  （ (日期,用户，广告) ，1）  （ (日期,用户，广告) ，1） ==>（ (日期,用户，广告) ，3）
        val redData = filterRdd.map(data => { //data ==>{ 16465487987,张三,加多宝，16465487987,张三,加多宝，16465487987,张三,加多宝， }
          val user = data.user //张三
          val ad = data.ad //,加多宝
          // data.ts ==时间戳 ==> yyyy-MM-dd 格式
          //时间格式转换
          val sdf = new SimpleDateFormat("yyyy-MM-dd")
          val day = sdf.format(new java.util.Date(data.ts.toLong)) //2023-4-3
          //（ (日期,用户，广告) ，1）  -->((2023-4-3,张三,加多宝),1)  (2023-4-3,张三,加多宝),1)  (2023-4-3,张三,加多宝),1)
          ( (day, user, ad) , 1)
          //  key             value
        })
        // ((2023-4-3,张三,加多宝),1)  (2023-4-3,张三,加多宝),1)  (2023-4-3,张三,加多宝),1)  => ((2023-4-3,张三,加多宝),3)
        val reduceData = redData.reduceByKey(_+_)
        reduceData
      })
  }

  //检查用户点击次数
  //如果 不满足拉入黑名单的广告条件 则存入每日用户点击广告次数表中，当存入表后满足拉入黑名单的条件则拉入黑名单
  private def checkUserCount(ds:DStream[((String, String, String), Int)]): Unit ={

    //DStream 没有输出可以执行吗？ 不能   输出有几种：print foreach
    //   ds 就是刚才所过滤 并 合并出来的数据
    ds.foreachRDD( rdd=>{
      rdd.foreach{
        case( (day,user,ad),count ) =>{
          println(s"${day},${user},${ad},${count}")
          //如果统计数据超过阈值(20),那么将用户拉入黑名单
          val blackListDao = new BlackListDao
          if(count >= 20){
            blackListDao.insertBlackList(user)//将超过20次的用户id 拉入黑名单
          }else{
            //否则放入 当天的点击统计次数表中
            blackListDao.insertUserAdCount(day,user,ad,count)
          }
          //判断更新后的点击数据是否超过阈值，如果超过，也拉黑
          val bool = blackListDao.checkUserByCount(day, user, ad, 20)
          if(bool){
            blackListDao.insertBlackList(user)
          }

        }
      }
    })
  }

}
